package com.thp.bigdata.logEnhance;

import java.io.IOException;

import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
 * 日志增强的自定义OutputFormat
 * 根据数据的不同输出类型到不同的输出目录
 * @author 汤小萌
 *
 */
public class LogEnhanceOutputFormat extends FileOutputFormat<Text, NullWritable> {

	@Override
	public RecordWriter<Text, NullWritable> getRecordWriter(TaskAttemptContext job) throws IOException, InterruptedException {
		FileSystem fs = FileSystem.get(job.getConfiguration());
		Path enhancePath = new Path("f:/enhancelog/output_en/log.txt");
		Path tocrawlPath = new Path("f:/enhancelog/output_crw/url.txt");
		FSDataOutputStream  enhanceOS = fs.create(enhancePath);
		FSDataOutputStream  tocrawlOS = fs.create(tocrawlPath);
		
		return new EnhanceRecordWriter(enhanceOS, tocrawlOS);
	}
	
	
	/**
	 * 这个RecordWriter类才是真正往外写文件的
	 * 需要往这个类的构造方法中传递输出流，在getRecordWriter()方法中就要进行构建这两个输出流
	 * @author 汤小萌
	 *
	 */
	static class EnhanceRecordWriter extends RecordWriter<Text, NullWritable> {
		FSDataOutputStream enhanceOS = null;
		FSDataOutputStream tocrawlOS = null;
		public EnhanceRecordWriter(FSDataOutputStream enhanceOS, FSDataOutputStream tocrawlOS) {
			super();
			this.enhanceOS = enhanceOS;
			this.tocrawlOS = tocrawlOS;
		}
		
		// 往外写文件的逻辑
		@Override
		public void write(Text key, NullWritable value) throws IOException, InterruptedException {
			String dataLine = key.toString(); 
			if(dataLine.contains("tocrawl")) {
				// 如果写的数据里面包含 "tocrawl"，那么就是不完全的数据，需要写入待爬清单文件
				tocrawlOS.write(dataLine.getBytes());
			} else {
				// 如果写的数据没有包含"tocrawl",就说明写出的数据是增强日志，那么就需要写入增强日志的文件
				enhanceOS.write(dataLine.getBytes());
			}
		}

		@Override
		public void close(TaskAttemptContext context) throws IOException, InterruptedException {
			if(enhanceOS != null) {
				enhanceOS.close();
			}
			if(tocrawlOS != null) {
				tocrawlOS.close();
			}
		}
		
	}

}
